package com.guanji.learn;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

public class FluxService {

    public static void main(String[] args) throws InterruptedException {

        // 来自搜索结果的核心示例[^17^]
        Flux<String> fruitFlux = Flux.just("Apple", "Banana", "Cherry");
        Flux<List<String>> buffer = fruitFlux.buffer(2);
        buffer.subscribe(fruit -> System.out.println("🍎 Fruit: " + fruit));

        System.out.println("------------------------");

        // 集合转化
        Flux<String> flux = Flux.fromIterable(Arrays.asList("Apple", "Banana", "Cherry"));
        Flux<Flux<String>> window = flux.window(2);
        window.subscribe(fruit -> {
            fruit.subscribe(System.out::println);
        });

        System.out.println("------------------------");

        // 整数序列
        Flux<Integer> integerFlux = Flux.range(1, 5);
        integerFlux.subscribe(System.out::println);

        System.out.println("------------------------");

        // 定时发送
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));
        interval.subscribe(fruit -> System.out.println("🍎 Fruit: "+ fruit) );

        System.out.println("------------------------");

        // 空结果
        Flux.empty().subscribe(System.out::println);

        // 过滤操作
        Flux<Integer> integerFlux1 = Flux.range(1, 50);
        integerFlux1.filter(i -> i <= 30)
                .distinct()
                .take(20)
                .skip(10)
                .elementAt(5)
                .subscribe(System.out::println);


        System.out.println("------------------------");
        // 输出: 0, 1, 2, 3, 4（5秒后自动停止）
        Flux.interval(Duration.ofSeconds(2), Duration.ofSeconds(1)) // 延迟2秒启动，之后每秒1次
                .take(5) // 只取前5个
                .subscribe(i -> System.out.println("Value: " + i));


        Thread.sleep(100000L);

        System.out.println("------------------------");

        Flux.interval(Duration.ofMillis(500))
                .publishOn(Schedulers.boundedElastic()) // 切换调度器
                .doOnNext(i -> System.out.println("Thread: " + Thread.currentThread().getName()))
                .take(3)
                .blockLast();

        System.out.println("------------------------");



    }

}
